package com.tensai.study.java.newconcept.typereactiveprogram.reactor.flux.create.statics.interval;

import cn.hutool.core.thread.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.ThreadLocalRandom;

/**
 * 定时器
 *
 * @author Tensai
 */
@Slf4j
public class IntervalTest {

	double max = 0;
	/**
	 * 一参测试
	 */
	@Test
	public void intervalTest1() {
		int t = Runtime.getRuntime().availableProcessors();
		log.info("可用处理器数量 - {}", t);
		// 五秒执行一次
		Flux.interval(Duration.ofNanos(1))
				.subscribe(this::sout);
		// hold main thread 让之前的异步线程能够一直跑
		ThreadUtil.sleep(5000);

	}

	/**
	 * 三参测试
	 */
	@Test
	public void intervalTest13() {
		Flux.interval(Duration.ofSeconds(2), Duration.ofSeconds(5), Schedulers.newSingle("device-session-checker"))
				.flatMap(i -> {
					log.info("tail- {} - {}", i, Instant.now().toEpochMilli());
					return Flux.empty();
				})
				.subscribe();
		ThreadUtil.sleep(5000);
	}

	private void sout(Long l) {

//		五秒执行一次 六秒睡眠 则任务间隔为六秒
		ThreadUtil.sleep(6000);
		double t = ThreadLocalRandom.current().nextGaussian();
		if (Math.abs(t) > max) {
			max = Math.abs(t);
			System.out.println(t + "  \t" + l + "\t" + max);
		}

	}

	/**
	 * 边界测试
	 * 不用测试了，有生之年不会越界
	 */
	@Test
	public void testBound() {
		Flux.interval(Duration.ofNanos(1))
				.subscribe(this::sout);
	}

	@Test
	public void csdn(){
		Flux.interval(Duration.ofMillis(200)).take(10)
				.subscribe(i -> System.out.println("  ->"+i + ":" + Instant.now().toString()));

		ThreadUtil.sleep(5000);

	}

	@Test
	public void count(){
		// Encode
		Disposable disposable = Flux.interval(Duration.ofNanos(200))
				.map(it ->CountEnum.INST.add())
				.subscribe();


		// startUp
		Flux.interval(Duration.ofSeconds(5))
				.map(it ->CountEnum.INST.get())
				.subscribe(i -> log.info(i + ""));

		ThreadUtil.sleep(50000);
		disposable.dispose();

	}
}
